// Notification Service - 通知服务

use axum::{
    extract::{
        ws::{Message, WebSocket, WebSocketUpgrade},
        Json, Path, Query, State,
    },
    http::StatusCode,
    response::IntoResponse,
    routing::{get, post, patch},
    Router,
};
use common::ApiResponse;
use serde::{Deserialize, Serialize};
use sqlx::PgPool;
use std::sync::Arc;
use tokio::sync::broadcast;
use auth::AuthUser;

// ========== 数据模型 ==========

#[derive(Debug, Serialize, Deserialize, sqlx::FromRow, Clone)]
struct Notification {
    id: String,
    user_id: String,
    #[serde(rename = "type")]
    notification_type: String,
    title: String,
    content: String,
    data: Option<serde_json::Value>,
    channel: String,
    status: String,
    sent_at: Option<chrono::DateTime<chrono::Utc>>,
    read_at: Option<chrono::DateTime<chrono::Utc>>,
    created_at: chrono::DateTime<chrono::Utc>,
}

#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
struct NotificationTemplate {
    id: String,
    #[serde(rename = "type")]
    template_type: String,
    title_template: String,
    content_template: String,
    channels: Vec<String>,
}

// ========== API 请求/响应 ==========

#[derive(Debug, Deserialize)]
struct SendNotificationRequest {
    user_id: String,
    notification_type: String,
    data: Option<serde_json::Value>,  // 模板变量
}

#[derive(Debug, Serialize)]
struct SendNotificationResponse {
    notification_id: String,
}

#[derive(Debug, Deserialize)]
struct ListQuery {
    status: Option<String>,
    page: Option<i64>,
    page_size: Option<i64>,
}

// ========== 应用状态 ==========

type NotificationBroadcaster = broadcast::Sender<Notification>;

#[derive(Clone)]
struct AppState {
    db: PgPool,
    broadcaster: NotificationBroadcaster,
}

// ========== 处理器 ==========

/// 健康检查
async fn health_check() -> (StatusCode, Json<ApiResponse<String>>) {
    (StatusCode::OK, Json(ApiResponse::ok("OK".to_string())))
}

/// WebSocket 连接处理
async fn websocket_handler(
    ws: WebSocketUpgrade,
    auth_user: AuthUser,
    State(state): State<Arc<AppState>>,
) -> impl IntoResponse {
    ws.on_upgrade(move |socket| handle_socket(socket, auth_user, state))
}

/// 处理 WebSocket 连接
async fn handle_socket(socket: WebSocket, auth_user: AuthUser, state: Arc<AppState>) {
    use futures::{SinkExt, StreamExt};
    let (mut sender, mut receiver) = socket.split();
    
    // 订阅广播频道
    let mut rx = state.broadcaster.subscribe();
    
    tracing::info!("WebSocket connected: user {}", auth_user.user_id);

    // 发送欢迎消息
    let welcome = serde_json::json!({
        "type": "connected",
        "user_id": auth_user.user_id,
        "message": "WebSocket 连接成功"
    });
    if sender.send(Message::Text(welcome.to_string())).await.is_err() {
        return;
    }

    // 双向通信
    let user_id = auth_user.user_id.clone();
    let send_task = tokio::spawn(async move {
        while let Ok(notification) = rx.recv().await {
            // 只发送给目标用户的通知
            if notification.user_id == user_id {
                let msg = serde_json::to_string(&notification).unwrap_or_default();
                if sender.send(Message::Text(msg)).await.is_err() {
                    break;
                }
            }
        }
    });

    let recv_task = tokio::spawn(async move {
        while let Some(Ok(msg)) = receiver.next().await {
            if let Message::Text(text) = msg {
                tracing::debug!("Received from client: {}", text);
                // 可以处理客户端发来的消息（如心跳）
            }
        }
    });

    // 等待任一任务结束
    tokio::select! {
        _ = send_task => {},
        _ = recv_task => {},
    }

    tracing::info!("WebSocket disconnected: user {}", auth_user.user_id);
}

/// 发送通知
async fn send_notification(
    _auth_user: AuthUser,  // 可以限制为系统内部调用
    State(state): State<Arc<AppState>>,
    Json(req): Json<SendNotificationRequest>,
) -> (StatusCode, Json<ApiResponse<SendNotificationResponse>>) {
    // 1. 获取通知模板
    let template: Option<NotificationTemplate> = sqlx::query_as(
        "SELECT * FROM notification_templates WHERE type = $1"
    )
    .bind(&req.notification_type)
    .fetch_optional(&state.db)
    .await
    .unwrap_or(None);

    let (title, content, channels) = if let Some(tmpl) = template {
        // 渲染模板（简化版 - 直接替换变量）
        let title = render_template(&tmpl.title_template, &req.data);
        let content = render_template(&tmpl.content_template, &req.data);
        (title, content, tmpl.channels)
    } else {
        // 使用默认值
        (
            format!("系统通知：{}", req.notification_type),
            "您有新的通知".to_string(),
            vec!["in_app".to_string()],
        )
    };

    // 2. 创建通知记录
    let notification_id = common::generate_id();
    
    for channel in &channels {
        let result = sqlx::query(
            r#"
            INSERT INTO notifications (id, user_id, type, title, content, data, channel)
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            "#
        )
        .bind(&notification_id)
        .bind(&req.user_id)
        .bind(&req.notification_type)
        .bind(&title)
        .bind(&content)
        .bind(&req.data)
        .bind(channel)
        .execute(&state.db)
        .await;

        if result.is_err() {
            tracing::error!("Failed to insert notification");
            continue;
        }

        // 3. 如果是 WebSocket 渠道，实时推送
        if channel == "websocket" || channel == "in_app" {
            let notification = Notification {
                id: notification_id.clone(),
                user_id: req.user_id.clone(),
                notification_type: req.notification_type.clone(),
                title: title.clone(),
                content: content.clone(),
                data: req.data.clone(),
                channel: channel.clone(),
                status: "sent".to_string(),
                sent_at: Some(chrono::Utc::now()),
                read_at: None,
                created_at: chrono::Utc::now(),
            };

            // 广播通知（忽略错误，因为用户可能不在线）
            let _ = state.broadcaster.send(notification);
        }
    }

    tracing::info!("Notification sent: {} to user {}", notification_id, req.user_id);

    (
        StatusCode::CREATED,
        Json(ApiResponse::ok(SendNotificationResponse { notification_id })),
    )
}

/// 获取用户通知列表
async fn list_notifications(
    auth_user: AuthUser,
    State(state): State<Arc<AppState>>,
    Query(query): Query<ListQuery>,
) -> (StatusCode, Json<ApiResponse<Vec<Notification>>>) {
    let page = query.page.unwrap_or(1).max(1);
    let page_size = query.page_size.unwrap_or(20).min(100);
    let offset = (page - 1) * page_size;

    let mut sql = "SELECT * FROM notifications WHERE user_id = $1".to_string();
    
    if let Some(status) = &query.status {
        sql.push_str(&format!(" AND status = '{}'", status));
    }
    
    sql.push_str(" ORDER BY created_at DESC LIMIT $2 OFFSET $3");

    let notifications = sqlx::query_as::<_, Notification>(&sql)
        .bind(&auth_user.user_id)
        .bind(page_size)
        .bind(offset)
        .fetch_all(&state.db)
        .await;

    match notifications {
        Ok(notifs) => (StatusCode::OK, Json(ApiResponse::ok(notifs))),
        Err(e) => {
            tracing::error!("Failed to fetch notifications: {}", e);
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(ApiResponse::error("查询失败".to_string())),
            )
        }
    }
}

/// 标记通知为已读
async fn mark_as_read(
    auth_user: AuthUser,
    State(state): State<Arc<AppState>>,
    Path(notification_id): Path<String>,
) -> (StatusCode, Json<ApiResponse<String>>) {
    let result = sqlx::query(
        "UPDATE notifications SET status = 'read', read_at = NOW() WHERE id = $1 AND user_id = $2"
    )
    .bind(&notification_id)
    .bind(&auth_user.user_id)
    .execute(&state.db)
    .await;

    match result {
        Ok(r) if r.rows_affected() > 0 => {
            (StatusCode::OK, Json(ApiResponse::ok("已标记为已读".to_string())))
        }
        Ok(_) => (
            StatusCode::NOT_FOUND,
            Json(ApiResponse::error("通知不存在".to_string())),
        ),
        Err(e) => {
            tracing::error!("Failed to mark as read: {}", e);
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(ApiResponse::error("操作失败".to_string())),
            )
        }
    }
}

/// 获取未读数量
async fn get_unread_count(
    auth_user: AuthUser,
    State(state): State<Arc<AppState>>,
) -> (StatusCode, Json<ApiResponse<i64>>) {
    let count: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM notifications WHERE user_id = $1 AND status = 'unread'"
    )
    .bind(&auth_user.user_id)
    .fetch_one(&state.db)
    .await
    .unwrap_or(0);

    (StatusCode::OK, Json(ApiResponse::ok(count)))
}

// ========== 辅助函数 ==========

/// 渲染通知模板（简单的变量替换）
fn render_template(template: &str, data: &Option<serde_json::Value>) -> String {
    let mut result = template.to_string();
    
    if let Some(data) = data {
        if let Some(obj) = data.as_object() {
            for (key, value) in obj {
                let placeholder = format!("{{{}}}", key);
                let replacement = match value {
                    serde_json::Value::String(s) => s.clone(),
                    serde_json::Value::Number(n) => n.to_string(),
                    _ => value.to_string(),
                };
                result = result.replace(&placeholder, &replacement);
            }
        }
    }
    
    result
}

// ========== 主函数 ==========

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    use common::service::*;
    
    init_tracing();
    dotenvy::dotenv().ok();

    let db = init_db("DATABASE_URL", "postgresql://intelligrade:password@localhost:5432/intelligrade").await?;
    let (broadcaster, _) = broadcast::channel::<Notification>(1000);

    let app = Router::new()
        .route("/health", get(health_check))
        .route("/ws", get(websocket_handler))
        .route("/api/v1/notifications", get(list_notifications))
        .route("/api/v1/notifications/send", post(send_notification))
        .route("/api/v1/notifications/:id/read", patch(mark_as_read))
        .route("/api/v1/notifications/unread-count", get(get_unread_count))
        .with_state(Arc::new(AppState { db, broadcaster }));

    start_service("Notification service", &env_or("NOTIFICATION_SERVICE_PORT", "8084"), app).await
}
